package day04

import java.net.URL

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

import scala.collection.mutable

object SubjectModuleCount2 {
  /**
    * 这次是带缓存的
    */
  def main(args: Array[String]): Unit = {
    /**
      * 需求：求各个学科各个模块的访问量的top3
      * 实现思路：
      *   1.计算出每个学科的各个模块（url）的访问数量
      *   2.按照学科进行分组
      *   3.组内排序并取的top3
      */

    val conf = new SparkConf().setAppName("SubjectModuleCount").setMaster("local[2]")
    val sc: SparkContext = new SparkContext(conf)
    val logs: RDD[String] = sc.textFile("D://视频资料/大数据/spark/Day04/access.txt")

    //数据：http://java.learn.com/java/javaee.shtml   注意学科是http://java.learn.com，后面的是模块
    //将用户访问日志进行切片并返回url
    val url: RDD[String] = logs.map(line => line.split("\t")(1))
    //将url生成元祖，便于聚合
    val tupUrl: RDD[(String, Int)] = url.map((_,1))
    //获取每个学科的各个模块的访问量
    /**
      * 从这里开始就有缓存了
      */
    //使用cache缓存的时候一定要注意一个问题：cache缓存的数据不要太大。
    //比如有4g的数据，你缓存了3g的数据到内存里，那肯定是不合适的。缓存的数据要适合节点的内存大小。
    val reducedUrl: RDD[(String, Int)] = tupUrl.reduceByKey(_+_).cache()
    //通过上面的数据来整合重新整合数据为二元组方便聚合：(学科subject,(url,count))

    //获取学科。
    val subjects: Array[String] = reducedUrl.map(x => {
      val arr: Array[String] = x._1.split("/")
      val subject: String = arr(0) + "//" + arr(2)
      subject
    }).distinct().collect()

    //用for循环对学科和数据进行匹配
    //注意java中有方法startsWith(...)可以匹配字符串开始字符
    for(subject <- subjects) {
      val value: RDD[(String, Int)] = reducedUrl.filter(_._1.startsWith(subject))
      val tuples: Array[(String, Int)] = value.sortBy(_._2, false).take(3)
      println(tuples.toBuffer)
    }

  }
}
